package task

import (
	"EasyReson/global"
	"database/sql"
	"EasyReson/sqls"
	log "code.google.com/p/log4go"
	"fmt"
	"strings"
	"EasyReson/util"
	"EasyReson/reson"
	"sync"
)

/**
	统计类型的sql语句中的关键字
 */
const count="count"

/**
	执行任务的单元
 */
type UnitTask struct {
	db1 *sql.DB //第一个数据源
	db2 *sql.DB //第二个数据源
	dbDest *sql.DB //目标数据源
	successResults []*ResonSuccess //勾兑上的数据
	reson1FailResults []*ResonFail //源1勾兑失败的数据
	reson2FailResults []*ResonFail //源2勾兑失败的数据
	task *global.Task //任务详情

	Running bool //当前任务是否在执行
	lock *sync.Mutex //程序执行锁
	Pause bool  //是否是暂停状态，如果状态是暂停，则逻辑不执行
}
/**
	数据成功勾兑上的
 */
type  ResonSuccess struct {
	db1Data map[string]interface{} //源1的数据
	db2Data map[string]interface{} //源2的数据
}
/**
	勾兑失败的对象
 */
type ResonFail struct {
	dbData map[string]interface{}
}

/**
	定时任务运行单元
	只有非暂停状态且上一个任务完成以后，才能触发下一个任务
 */
func (t *UnitTask) cronRun() {
	if !t.Pause {
		if !t.Running && t.setRunning() {
			t.cronTaskRun()
			defer t.setUnRunning()
		}
	}else{
		log.Info("taskId [%s] taskName [%s] status is stopped now",
			t.task.TaskId,t.task.TaskName)
	}
}
/**
	真正触发定时任务
 */
func (t *UnitTask) cronTaskRun() {
	/**
		源数据库的BeforeSql 前置条件检查
	 */
	if !t.srcDbExecBeforeSql(){
		log.Info("taskId [%s] taskName [%s] has BeforeSql result/count is 0, job excuseDown [%s]",
			t.task.TaskId,t.task.TaskName,util.GetCurrentTimeStr())
		return
	}
	/**
		勾兑差异数据库的BeforeSql 前置sql语句执行,如果执行出错，本次任务也终止
	 */
	 err := t.destDbExecBeforeSql()
	 if err!= nil {
	 	errStr := fmt.Sprintf("taskId [%s] taskName [%s] [%s] BeforeSql Excuse Error [%s], job excuseDown [%s]",
			t.task.TaskId,t.task.TaskName,"db-dest",err,util.GetCurrentTimeStr())
		 log.Info(errStr)
		 return
	 }
	 //查询数据集
	src1Results,src2Results :=reson.GetSrcResult(t.task.Db1Src,t.task.Db2Src,t.db1,t.db2,t.task.TaskId,t.task.TaskName)
	if len(src1Results)==0||len(src2Results)==0 {
		errStr := fmt.Sprintf("taskId [%s] taskName [%s] have emptyResults，src1Results size [%v], src2Results[%v], job excuseDown [%s]",
			t.task.TaskId,t.task.TaskName,len(src1Results),len(src2Results),util.GetCurrentTimeStr())
		log.Info(errStr)
		return
	}
	src1Map := make(map[string]map[string]interface{})
	for i:= range src1Results {
		src1Map[t.buildJoinKey(src1Results[i],"1")]=src1Results[i]
	}
	//将src1的数据方入map以后，显式释放Src1的数据集切片对象，防止占用大的内存
	src1Results=nil

	//勾兑成功的数据
	successResults := make([]*ResonSuccess,0)
	//存储源1勾兑的失败的数据：
	reson1FailResults := make([]*ResonFail,0)
	//存储源2勾兑的失败的数据：
	reson2FailResults := make([]*ResonFail,0)
	for j:= range src2Results {
		src2JoinKey := t.buildJoinKey(src2Results[j],"2")

		//勾兑成功的数据，勾兑成功后要删除两边源头的数据
		if _,ok := src1Map[src2JoinKey];ok {
			resonSuccess := &ResonSuccess{src1Map[src2JoinKey],src2Results[j]}
			successResults =append(successResults,resonSuccess)
			delete(src1Map,src2JoinKey)
		}else{
			//表2中有，表1中没有
			reson2Fail := &ResonFail{src2Results[j]}
			reson2FailResults=append(reson2FailResults,reson2Fail)
		}
	}
	//表1中有，表2中没有
	for _,v := range src1Map{
		reson1Fail := &ResonFail{v}
		reson1FailResults=append(reson1FailResults,reson1Fail)
	}
	//数据清清洗完毕，将map和src2对象置空，防止大对象占用内存
	src2Results = nil
	src1Map= nil

	t.successResults = successResults
	t.reson1FailResults=reson1FailResults
	t.reson2FailResults=reson2FailResults

	log.Info("taskId [%s] taskName [%s] reson Success size [%v] ; reson db1-src fail size [%v];reson db2-src fail size [%v] ",
		t.task.TaskId,t.task.TaskName,len(successResults),len(reson1FailResults),len(reson2FailResults))

	t.reFreshData()

	log.Info("taskId [%s] taskName [%s] reson is ok! job excuseDown [%s]",
		t.task.TaskId,t.task.TaskName,util.GetCurrentTimeStr())

}
/**
	如果db1和db2有配置beforesql，则执行
 */
func (t *UnitTask) srcDbExecBeforeSql() bool{

	beforeTasks := make(map[string]*global.DbSrc)
	beforeTasks["db1-src"]=t.task.Db1Src
	beforeTasks["db2-src"]=t.task.Db2Src

	dbTasks := make(map[string]*sql.DB)
	dbTasks["db1-src"]=t.db1
	dbTasks["db2-src"]=t.db2

	for k,v := range beforeTasks {
		if "" == v.BeforeSql {
			continue
		}
		results, err := sqls.GetList(v.BeforeSql, dbTasks[k])
		if err != nil {
			errStr := fmt.Sprintf("taskId [%s] taskName [%s] [%s] beforeSql error [%s]",
				t.task.TaskId, t.task.TaskName, k, err)
			log.Error(errStr)
			return false

		}
		//非统计查询
		if !strings.Contains(v.BeforeSql, count) {
			log.Info("taskId [%s] taskName [%s] [%s] BeforeSql result size [%v]",
				t.task.TaskId, t.task.TaskName, k, len(results))
			if len(results) == 0 {
				return false
			}
			//统计查询
		} else {
			for _, count := range results[0] {
				log.Info("taskId [%s] taskName [%s] [%s] BeforeSql result count [%v]",
					t.task.TaskId, t.task.TaskName, k, count)
				if count == "0" {
					return false
				}
			}
		}
	}
	return true
}
/**
	勾兑差异目标库的前置sql执行(可能是删或者改，但不能是truncate)
 */
func (t *UnitTask) destDbExecBeforeSql() error{

	sqlStr := t.task.DBDest.BeforeSql
	affected,err := sqls.ExecSql(sqlStr,t.dbDest,nil)
	log.Info("taskId [%s] taskName [%s] [%s] BeforeSql RowsAffected [%v]",
		t.task.TaskId,t.task.TaskName,"db-dest",affected)
	if err!= nil {
		return err
	}
	return nil
}
/**
	建立JoinKey
 */
func (t *UnitTask) buildJoinKey(src map[string]interface{}, index string) string{
	//将数据的key全部转化为小写，好进行大小写兼容
	srcToLower := make(map[string]interface{})
	for k,v := range src{
		srcToLower[util.ToLowerCase(k)]=v
	}

	joinKeys := t.task.JoinKey
	joinKeyList := joinKeys[index]

	result := ""
	for i := range joinKeyList{
		joinKey := util.ToLowerCase(joinKeyList[i])
		result = result+util.ToString(srcToLower[joinKey])+"_"
	}
	result=result[0:len(result)-1]
	return result
}

/**
	进行数据回写或者数据清洗
	分3个协程池进行数据处理，
	勾兑成功的协程池和勾兑失败的2个协程池所占用的总数可以是18：1：1
 */
func (t *UnitTask) reFreshData() {

	thread := t.task.Thread
	success:=util.GetIntByMulTiFloat(thread,0.95)
	src1Fail := util.GetIntByMulTiFloat(thread,0.05)
	src2Fail := util.GetIntByMulTiFloat(thread,0.05)

	successUnit := &successUnit{t,success,100,nil}
	successUnit.initWorkerPool()
	successUnit.submitSuccessTask(t.successResults)


	src1FailUnit := &srcFailUnit{t,src1Fail,100,nil,1}
	src1FailUnit.initWorkerPool()
	src1FailUnit.submitFailTask(t.reson1FailResults)

	src2FailUnit := &srcFailUnit{t,src2Fail,100,nil,2}
	src2FailUnit.initWorkerPool()
	src2FailUnit.submitFailTask(t.reson2FailResults)
}


/**
	将未执行状态改为执行状态
	如果程序正在执行，更改则失败。利用互斥锁实现，仅对当前结构体的多次调用生效
 */
func (t *UnitTask) setRunning() (success bool ){
	t.lock.Lock()
	if t.Running {
		return false
	}
	t.Running =true
	defer t.lock.Unlock()
	return true
}
/**
	将正在执行的任务的状态改为未执行
 */
func (t *UnitTask) setUnRunning() {
	t.Running =false
}